[小ネタ]バックアップファイルがS3バケットに保存されているか定期的に確認してみた

荒平祐次(arap)

2023.11.20

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、AWS事業本部の荒平(@0Air)です。

ある製品にて、データをS3バケットに定期的にバックアップする必要があり、そのバックアップが取れているか確認して通知したいという要望がありました。

AWS Backupであれば、成功・失敗の通知は機能として備わっているのですが（下記参照）、S3ではオブジェクトが配置されるトリガー以外にイベントを発生させることができません。

今回は、EventBridgeにて定期的にS3バケットを確認し、タイムスタンプを見て直近のバックアップファイルが存在するかを確認する仕組みを作ってみました。

構成図

シンプルですが、以下のような構成を考えました。EventBridgeによりスケジュールし、LambdaにてS3バケットの内容を確認します。確認結果をAmazon SNSに送信します。
バックアップする部分については実装対象外です。

やってみた

1. S3バケット/SNSトピックを作成

バックアップ保存先のS3バケット、通知用のSNSトピックをそれぞれ作成します。
設定値は名称やメール送信先以外デフォルトで問題ありませんので割愛します。

2. Lambdaの作成

以下のLambdaをPython 3.x系のランタイムにて作成します。
sns_topic_arnおよび、bucket_nameは適宜、環境に合わせて置き換えます。

LastModified(最終変更時間)が3600秒を切っているオブジェクトがあれば「Success」、なければ「Failed」を返すようにしています。

import boto3
import datetime

def lambda_handler(event, context):
    # 環境に応じて置き換え
    sns_topic_arn = 'arn:aws:sns:ap-northeast-1:account-id:arap-topic-name'
    bucket_name = 'backup-data-arap'

    s3 = boto3.client('s3')
    sns = boto3.client('sns')
    now = datetime.datetime.now(datetime.timezone.utc)

    response = s3.list_objects_v2(Bucket=bucket_name)
    message = "S3 Backup Status: Failed"

    for obj in response.get('Contents', []):
        if (now - obj['LastModified']).total_seconds() < 3600:
            message = "S3 Backup Status: Success"
            break

    sns.publish(
        TopicArn=sns_topic_arn,
        Message=message,
        Subject="S3 Backup Status"
    )

    return message

Lambdaの実行ロールには、AmazonS3ReadOnlyAccessおよび、AmazonSNSFullAccessをアタッチします。

定期的にチェックしたいため、トリガーを追加します。Lambda画面にて、「トリガーを追加」をクリックします。

トリガーはEventBridge (CloudWatch Events)を選択します。
既存でルールがある場合はそれを選択し、新規でルール作成する場合はルール名、スケジュール式を入力し「追加」します。
検証では、1日待つわけには行かないので1時間ごとのトリガーを仕掛けていますが、一般的には24時間間隔など、バックアップの時間によって調整するとよいと思います。